package io.kiki.sba.registry.server.clientInterface.push;

import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import static io.kiki.sba.registry.server.clientInterface.push.PushMetrics.Push.PUSH_REG_COMMIT_COUNTER;
import static io.kiki.sba.registry.server.clientInterface.push.PushMetrics.Push.PUSH_REG_SKIP_COUNTER;

public final class RegProcessor {
    private static final Logger logger = LoggerFactory.getLogger(RegProcessor.class);
    final Worker[] workers;
    final RegHandler regHandler;
    private PushEfficiencyImproveConfig pushEfficiencyImproveConfig;

    RegProcessor(int workerSize, RegHandler regHandler) {
        this.regHandler = regHandler;
        this.workers = new Worker[workerSize];
        for (int i = 0; i < workers.length; i++) {
            Worker worker = new Worker();
            this.workers[i] = worker;
            ConcurrentUtils.createDaemonThread("RegProcessor-" + i, worker).start();
        }
    }

    public void setWorkDelayTime(PushEfficiencyImproveConfig pushEfficiencyImproveConfig) {
        this.pushEfficiencyImproveConfig = pushEfficiencyImproveConfig;
        for (Worker worker : this.workers) {
            worker.setRegWorkWaitingMillis(pushEfficiencyImproveConfig.getRegWorkWaitingMillis());
        }
    }

    boolean fireOnReg(Subscriber subscriber) {
        final String dataInfoId = subscriber.getDataInfoId();
        Worker worker = indexOf(dataInfoId);
        Buffer buffer = worker.map.computeIfAbsent(dataInfoId, k -> new Buffer());
        boolean result = buffer.add(subscriber);
        if (null != pushEfficiencyImproveConfig && pushEfficiencyImproveConfig.fetchRegWorkWake(subscriber.getAppName())) {
            worker.wakeup();
        }
        return result;
    }

    private Worker indexOf(String dataInfoId) {
        int n = (dataInfoId.hashCode() & 0x7fffffff) % workers.length;
        return workers[n];
    }

    int processBuffer(Ref ref, int hitSize) {
        List<Subscriber> list = new ArrayList<>(hitSize);
        for (Map.Entry<String, Subscriber> entry : ref.subscriberMap.entrySet()) {
            final Subscriber subscriber = entry.getValue();
            if (!subscriber.hasPushed()) {
                list.add(subscriber);
            }
            // try to remove the sub, but subs maybe changes
            ref.subscriberMap.remove(subscriber.getRegisterId(), subscriber);
        }
        if (!list.isEmpty()) {
            regHandler.onReg(ref.dataInfoId, list);
        }
        return list.size();
    }


    void suspend() {
        for (Worker worker : workers) {
            worker.suspend();
        }
    }

    private static final class Buffer {
        volatile Map<String, Subscriber> map = new ConcurrentHashMap<>(256);

        boolean add(Subscriber subscriber) {
            // the sub maybe replaced when adding
            final Map<String, Subscriber> map = this.map;
            final String registerId = subscriber.getRegisterId();
            // quick path
            if (map.putIfAbsent(registerId, subscriber) == null) {
                PUSH_REG_COMMIT_COUNTER.inc();
                return true;
            }
            for (; ; ) {
                final Subscriber existing = map.get(registerId);
                if (existing == null) {
                    if (map.putIfAbsent(registerId, subscriber) == null) {
                        PUSH_REG_COMMIT_COUNTER.inc();
                        return true;
                    }
                } else {
                    if (!existing.registerVersion().orderThan(subscriber.registerVersion())) {
                        PUSH_REG_SKIP_COUNTER.inc();
                        logger.info("[SkipReg]{}, existing={}, add={}", subscriber.getDataInfoId(), existing.registerVersion(), subscriber.shortDesc());
                        return false;
                    }
                    if (map.replace(registerId, existing, subscriber)) {
                        PUSH_REG_COMMIT_COUNTER.inc();
                        return true;
                    }
                }
            }
        }

        /**
         * 这里会重新new一个map
         *
         * @return
         */
        Map<String, Subscriber> getMap() {
            // For performance, just replace the subs
            // In a concurrent scenario, small probability there are still write operations on the
            // replaced subs
            if (map.isEmpty()) {
                return Collections.emptyMap();
            }
            Map<String, Subscriber> map = this.map;
            this.map = new ConcurrentHashMap<>(256);
            return map;
        }
    }

    static final class Ref {
        final String dataInfoId;
        // ref of BufferWorker.subMap, do not copy it
        final Map<String, Subscriber> subscriberMap;

        Ref(String dataInfoId, Map<String, Subscriber> subscriberMap) {
            this.dataInfoId = dataInfoId;
            this.subscriberMap = subscriberMap;
        }

        int size() {
            return subscriberMap.size();
        }
    }

    final class Worker extends WakeUpLoopExecuteTask {
        final Map<String, Buffer> map = new ConcurrentHashMap<>(1024 * 8);

        @Setter
        private int regWorkWaitingMillis = 200;

        @Override
        public void _execute_() {
            watchBuffer();
        }

        @Override
        public int getWaitingMillis() {
            return regWorkWaitingMillis;
        }

        int watchBuffer() {
            final List<Ref> list = getList();
            if (list.isEmpty()) {
                return 0;
            }
            int firstCount = 0;
            int secondCount = 0;
            int subSize = 0;
            for (Ref ref : list) {
                final int size = ref.size();
                if (size != 0) {
                    subSize += size;
                    firstCount += processBuffer(ref, size);
                }
            }
            // double process, maybe the sub.ref has update after first process
            ConcurrentUtils.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
            for (Ref ref : list) {
                final int size = ref.size();
                if (size != 0) {
                    subSize += size;
                    secondCount += processBuffer(ref, size);
                }
            }
            logger.info("[regBuffer]dataIds={},subs={},regs={},{},{}", list.size(), subSize, firstCount + secondCount, firstCount, secondCount);
            return firstCount + secondCount;
        }

        List<Ref> getList() {
            List<Ref> list = new ArrayList<>(1024);
            for (Map.Entry<String, Buffer> entry : this.map.entrySet()) {
                Map<String, Subscriber> map = entry.getValue().getMap();
                if (!map.isEmpty()) {
                    Ref ref = new Ref(entry.getKey(), map);
                    list.add(ref);
                }
            }
            return list;
        }
    }
}
